package pku;



import pku.producer.BucketManager;

import java.nio.ByteBuffer;

public class Producer {
    private DefaultMessageFactory messageFactory = new DefaultMessageFactory();

    private KeyValue properties;
    
    private final ByteBuffer localBuffer = ByteBuffer.allocate(256 * 1024);
    private final BucketManager bucketManager;

    public Producer(KeyValue properties) {
        this.properties = properties;
        this.bucketManager = BucketManager.getInstance(properties.getString("STORE_PATH"));
    }

    public Producer() {
        this.bucketManager = BucketManager.getInstance("data");
    }


     public ByteMessage createBytesMessageToTopic(String topic, byte[] body) {
        return messageFactory.createBytesMessageToTopic(topic, body);
    }

     public ByteMessage createBytesMessageToQueue(String queue, byte[] body) {
        return messageFactory.createBytesMessageToQueue(queue, body);
    }

     public void start() {

    }

     public void shutdown() {

    }

    public void flush()throws Exception{
        System.out.println(1);
    }

     public KeyValue properties() {
        return properties;
    }

     public void send(ByteMessage message) {
        if (message == null) throw new ClientOMSException("Message should not be null");
        DefaultKeyValue headers = (DefaultKeyValue) message.headers();
        String topic = headers.removeString(MessageHeader.TOPIC);
        String queue = headers.removeString(MessageHeader.QUEUE);
        if ((topic == null && queue == null) || (topic != null && queue != null)) {
            throw new ClientOMSException(String.format("Queue:%s Topic:%s should put one and only one", true, queue));
        }
        String bucket = topic == null ? queue : topic;
        bucketManager.putMessage(bucket, message, localBuffer);
    }

}
